package net.sf.nlp.ltag.parser.dot;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import net.sf.nlp.ltag.data.LTAGTree;
import net.sf.nlp.ltag.parser.LTAGState;
import net.sf.nlp.ltag.parser.ParserTree;

public class DotAlgMultiThread extends DotAlgAbstract
{
	protected static Logger logger =  Logger.getLogger(DotAlgMultiThread.class);
	
	private BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
	protected ConcurrentLinkedQueue<LTAGState> stateQueue = new ConcurrentLinkedQueue<LTAGState>();
	private ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 10000, TimeUnit.SECONDS, queue);
	
	private Object wait = new Object();
	
	public DotAlgMultiThread(Map<String, List<LTAGTree>> ltagMap)
	{
		super(ltagMap);
		
		new PollingStateQueueThread().start();
	}
	
	private void waitForFinish() throws InterruptedException
	{
		synchronized(wait)
		{
			wait.wait();
		}
	}
	
	public List<ParserTree> calculate() throws Exception
	{
		matchedStates.clear();
		
		List<LTAGState> stateSet = getInitStateSet();
		
		stateQueue.addAll(stateSet);
		
		waitForFinish();
		
		List<ParserTree> parserTrees = getParsedTree(matchedStates);
		
		return parserTrees;
	}
	
	public class PollingStateQueueThread extends Thread
	{
		private long taskCount = 0;
		private long computingTaskCount = 0;
		private List<Future<Boolean>> futureList = new LinkedList<Future<Boolean>>();
		
		public void run() 
		{
			while (true)
			{
				try
				{
					LTAGState state = stateQueue.poll();
					
					if (state != null)
					{
						taskCount++;
						computingTaskCount++;
						Future<Boolean> future = executor.submit(new CalculationThread(state));
						futureList.add(future);
					}
					else
					{
						if (isFinished())
						{
							synchronized(wait)
							{
								wait.notify();
							}
							
							logger.info("taskCount=" + taskCount);
						}
						
						Thread.sleep(5);
					}
				}
				catch(Exception ex)
				{
					logger.error(ex, ex);
				}
			}
		}
		
		private boolean isFinished()
		{
			Iterator<Future<Boolean>> i = futureList.iterator();
			Future<Boolean> future;
			
			while (i.hasNext())
			{
				future = i.next();
				
				if (!future.isDone())
					continue;
				
				computingTaskCount--;
				i.remove();
			}
			
			if (computingTaskCount == 0 && stateQueue.size() == 0)
				return true;
			
			return false;
		}
	}
	
	public class CalculationThread implements Callable<Boolean>
	{
		private List<LTAGState> stateSet = new ArrayList<LTAGState>();
		
		public CalculationThread(LTAGState state)
		{
			stateSet.add(state);
		}
		
		public Boolean call() throws Exception 
		{
			calculateTask(stateSet);
			
			return true;
		}
	}
}
